package com.bestvike.stone.spring.stream;

import com.bestvike.stone.spring.rabbit.RabbitAutoConfiguration;
import com.bestvike.stone.spring.stream.core.StreamRabbitMessageChannelBinder;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binder.rabbit.config.RabbitMessageChannelBinderConfiguration;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties;
import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

/**
 * Created by 许崇雷 on 2018-09-03.
 */
@Configuration
@Import({PropertyPlaceholderAutoConfiguration.class})
@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class})
@AutoConfigureAfter(RabbitAutoConfiguration.class)
@AutoConfigureBefore(RabbitMessageChannelBinderConfiguration.class)
@ConditionalOnClass(RabbitMessageChannelBinderConfiguration.class)
public class StreamRabbitMessageChannelBinderConfiguration {
    private final RabbitTemplate rabbitTemplate;

    private final ConnectionFactory rabbitConnectionFactory;

    private final RabbitProperties rabbitProperties;

    private final RabbitBinderConfigurationProperties rabbitBinderConfigurationProperties;

    private final RabbitExtendedBindingProperties rabbitExtendedBindingProperties;

    public StreamRabbitMessageChannelBinderConfiguration(RabbitTemplate rabbitTemplate,
                                                         ConnectionFactory rabbitConnectionFactory,
                                                         RabbitProperties rabbitProperties,
                                                         RabbitBinderConfigurationProperties rabbitBinderConfigurationProperties,
                                                         RabbitExtendedBindingProperties rabbitExtendedBindingProperties) {
        this.rabbitTemplate = rabbitTemplate;
        this.rabbitConnectionFactory = rabbitConnectionFactory;
        this.rabbitProperties = rabbitProperties;
        this.rabbitBinderConfigurationProperties = rabbitBinderConfigurationProperties;
        this.rabbitExtendedBindingProperties = rabbitExtendedBindingProperties;
    }

    @Bean
    RabbitMessageChannelBinder rabbitMessageChannelBinder() {
        RabbitMessageChannelBinder binder = new StreamRabbitMessageChannelBinder(this.rabbitTemplate, this.rabbitConnectionFactory, this.rabbitProperties, this.provisioningProvider());
        binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses());
        binder.setCompressingPostProcessor(this.gZipPostProcessor());
        binder.setDecompressingPostProcessor(this.deCompressingPostProcessor());
        binder.setNodes(this.rabbitBinderConfigurationProperties.getNodes());
        binder.setExtendedBindingProperties(this.rabbitExtendedBindingProperties);
        return binder;
    }

    @Bean
    MessagePostProcessor deCompressingPostProcessor() {
        return new DelegatingDecompressingPostProcessor();
    }

    @Bean
    MessagePostProcessor gZipPostProcessor() {
        GZipPostProcessor gZipPostProcessor = new GZipPostProcessor();
        gZipPostProcessor.setLevel(this.rabbitBinderConfigurationProperties.getCompressionLevel());
        return gZipPostProcessor;
    }

    @Bean
    RabbitExchangeQueueProvisioner provisioningProvider() {
        return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory);
    }
}
